package app;

import bean.OrderDetail;
import bean.OrderInfo;
import bean.OrderWide;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.HasOffsetRanges;
import org.apache.spark.streaming.kafka010.OffsetRange;
import redis.clients.jedis.Jedis;
import scala.Tuple2;
import util.MyEsUtils;
import util.MyKafkaUtils;
import util.MyOffsetsUtils;
import util.MyRedisUtils;

import java.time.LocalDate;
import java.time.Period;
import java.util.*;

/**
 * 订单宽表任务
 *
 * 1. 准备实时环境
 * 2. 从Redis中读取offset  * 2
 * 3. 从kakfa中消费数据 * 2
 * 4. 提取offset * 2
 * 5. 数据处理
 *    5.1 转换结构
 *    5.2 维度关联
 *    5.3 双流join
 * 6. 写入ES
 * 7. 提交offset * 2
 */
public class DwdOrderApp {
    private static final String orderInfoTopicName = "DWD_ORDER_INFO_I_1018";
    private static final  String orderInfoGroup = "DWD_ORDER_INFO:GROUP";

    private static final String orderDetailTopicName = "DWD_ORDER_DETAIL_I_1018";

    private static final  String orderDetailGroup = "DWD_ORDER_DETAIL_GROUP";

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("ods_order_app");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
        //2.读取offset
        //order_info
        Map<TopicPartition, Long> orderInfoOffsets = MyOffsetsUtils.readOffset(orderInfoTopicName, orderInfoGroup);
        //order_detail
        Map<TopicPartition, Long> orderDetailOffsets = MyOffsetsUtils.readOffset(orderInfoTopicName, orderInfoGroup);
        //3. 从Kafka中消费数据
        // order_info
        JavaInputDStream<ConsumerRecord<Object, Object>> orderInfoKafkaDStream = null;
        if(orderInfoOffsets != null && !orderInfoOffsets.isEmpty()) {
            orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(jssc,orderInfoTopicName, orderInfoGroup,orderInfoOffsets);
        } else {
            orderInfoKafkaDStream = MyKafkaUtils.getKafkaDStream(jssc,orderInfoTopicName, orderInfoGroup);
        }
        //order_detail
        JavaInputDStream<ConsumerRecord<Object, Object>> orderDetailKafkaDStream = null;
        if(orderDetailOffsets != null && !orderDetailOffsets.isEmpty()) {
            orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(jssc,orderDetailTopicName,orderDetailGroup,orderDetailOffsets);
        } else {
            orderDetailKafkaDStream = MyKafkaUtils.getKafkaDStream(jssc,orderDetailTopicName,orderDetailGroup);
        }
        final List<OffsetRange> orderInfoOffsetRanges = new ArrayList<OffsetRange>();
        // order_info
        JavaDStream<ConsumerRecord<Object, Object>> orderInfoOffsetDStream = orderInfoKafkaDStream .transform(new Function<JavaRDD<ConsumerRecord<Object, Object>>, JavaRDD<ConsumerRecord<Object, Object>>>() {
            public JavaRDD<ConsumerRecord<Object, Object>> call(JavaRDD<ConsumerRecord<Object, Object>> rdd) throws Exception {
                OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
                for(int i = 0; i < offsetRanges.length; i++) {
                    orderInfoOffsetRanges.add(offsetRanges[i]);
                }
                return rdd;
            }
        });

        //order_detail
        final List<OffsetRange> orderDetailOffsetRanges = new ArrayList<OffsetRange>();
        JavaDStream<ConsumerRecord<Object, Object>> orderDetailOffsetDStream = orderDetailKafkaDStream.transform(new Function<JavaRDD<ConsumerRecord<Object, Object>>, JavaRDD<ConsumerRecord<Object, Object>>>() {
            public JavaRDD<ConsumerRecord<Object, Object>> call(JavaRDD<ConsumerRecord<Object, Object>> rdd) throws Exception {
                OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
                for(int i = 0; i < offsetRanges.length; i++) {
                    orderDetailOffsetRanges.add(offsetRanges[i]);
                }
                return rdd;
            }
        });
        //5. 处理数据
        // 5.1 转换结构
        JavaDStream<OrderInfo> orderInfoDStream = orderInfoOffsetDStream.map(new Function<ConsumerRecord<Object, Object>, OrderInfo>() {
            @Override
            public OrderInfo call(ConsumerRecord<Object, Object> objectObjectConsumerRecord) throws Exception {
                String value = objectObjectConsumerRecord.value().toString();
                OrderInfo orderInfo = JSON.parseObject(value, OrderInfo.class);
                return orderInfo;
            }
        });

        JavaDStream<OrderDetail> orderDetailDStream = orderDetailOffsetDStream.map(new Function<ConsumerRecord<Object, Object>, OrderDetail>() {
            @Override
            public OrderDetail call(ConsumerRecord<Object, Object> objectObjectConsumerRecord) throws Exception {
                String value = objectObjectConsumerRecord.value().toString();
                OrderDetail orderDetail = JSON.parseObject(value, OrderDetail.class);
                return orderDetail;
            }
        });

        //5.2 维度关联
        // order_info
        JavaDStream<OrderInfo> orderInfoDimDStream = orderInfoDStream.mapPartitions(new FlatMapFunction<Iterator<OrderInfo>, OrderInfo>() {
            @Override
            public Iterator<OrderInfo> call(Iterator<OrderInfo> orderInfoIterator) throws Exception {
                List<OrderInfo> orderInfos = new ArrayList<>();
                Jedis jedis = MyRedisUtils.getJedisFromPool();
                while(orderInfoIterator.hasNext()) {
                    OrderInfo orderInfo = orderInfoIterator.next();
                    Long uid = orderInfo.getUser_id();
                    String redisUserKey = "DIM:USER_INFO:"+ uid;
                    String userInfoJson = jedis.get(redisUserKey);
                    JSONObject userInfoJsonObj = JSON.parseObject(userInfoJson);
                    //提取性别
                    String gender = userInfoJsonObj.getString("gender");
                    //提取生日
                    String birthday = userInfoJsonObj.getString("birthday");
                    //换算年龄
                    LocalDate birthdayLd = LocalDate.parse(birthday);
                    LocalDate nowLd = LocalDate.now();
                    Period  period = Period.between(birthdayLd, nowLd);
                    int age = period.getYears();
                    //补充到对象中
                    orderInfo.setUser_gender(gender);
                    orderInfo.setUser_age(age);
                    //关联地区维度
                    Long provinceID = orderInfo.getProvince_id();
                    String redisProvinceKey = "DIM:BASE_PROVINCE:" + provinceID;
                    String provinceJson = jedis.get(redisProvinceKey);
                    JSONObject provinceJsonObj = JSON.parseObject(provinceJson);
                    String provinceName = provinceJsonObj.getString("name");
                    String provinceAreaCode = provinceJsonObj.getString("area_code");
                    String province3166 = provinceJsonObj.getString("iso_3166_2");
                    String provinceIsoCode = provinceJsonObj.getString("iso_code");
                    //补充到对象中
                    orderInfo.setProvince_name(provinceName);
                    orderInfo.setProvince_area_code(provinceAreaCode);
                    orderInfo.setProvince_3166_2_code(province3166);
                    orderInfo.setProvince_iso_code(provinceIsoCode);
                    //处理日期字段
                    String createTime = orderInfo.getCreate_time();
                    String[] createDtHr = createTime.split(" ");
                    String createDate = createDtHr[0];
                    String createHr = (createDtHr[1].split(":"))[0];
                    //补充到对象中
                    orderInfo.setCreate_date(createTime);
                    orderInfo.setCreate_hour(createHr);
                    orderInfos.add(orderInfo);
                }
                jedis.close();
                return orderInfos.iterator();
            }
        });
        //5.3 双流Join
        // 内连接 join  结果集取交集
        // 外连接
        //   左外连  leftOuterJoin   左表所有+右表的匹配  , 分析清楚主(驱动)表 从(匹配表) 表
        //   右外连  rightOuterJoin  左表的匹配 + 右表的所有,分析清楚主(驱动)表 从(匹配表) 表
        //   全外连  fullOuterJoin   两张表的所有

        // 从数据库层面： order_info 表中的数据 和 order_detail表中的数据一定能关联成功.
        // 从流处理层面:  order_info 和  order_detail是两个流， 流的join只能是同一个批次的数据才能进行join
        //               如果两个表的数据进入到不同批次中， 就会join不成功.
        // 数据延迟导致的数据没有进入到同一个批次，在实时处理中是正常现象. 我们可以接收因为延迟导致最终的结果延迟.
        // 我们不能接收因为延迟导致的数据丢失.
        JavaPairDStream<Long, OrderInfo> orderInfoKVDStream = orderInfoDimDStream.mapToPair(new PairFunction<OrderInfo, Long, OrderInfo>() {
            @Override
            public Tuple2<Long, OrderInfo> call(OrderInfo orderInfo) throws Exception {
                return new Tuple2<Long, OrderInfo>(orderInfo.getId(),orderInfo);
            }
        });

        JavaPairDStream<Long, OrderDetail> orderDetailKVDStream = orderDetailDStream.mapToPair(new PairFunction<OrderDetail, Long, OrderDetail>() {
            @Override
            public Tuple2<Long, OrderDetail> call(OrderDetail orderDetail) throws Exception {
                return new Tuple2<Long, OrderDetail>(orderDetail.getOrder_id(),orderDetail);
            }
        });

        // 解决:
        //  1. 扩大采集周期 ， 治标不治本
        //  2. 使用窗口,治标不治本 , 还要考虑数据去重 、 Spark状态的缺点
        //  3. 首先使用fullOuterJoin,保证join成功或者没有成功的数据都出现到结果中.
        //     让双方都多两步操作, 到缓存中找对的人， 把自己写到缓存中
        //JavaDStream<Long, Tuple2<Optional<OrderInfo>, Optional<OrderDetail>>> orderJoinDStream

        JavaPairDStream<Long, Tuple2<Optional<OrderInfo>, Optional<OrderDetail>>> orderJoinDStream = orderInfoKVDStream.fullOuterJoin(orderDetailKVDStream);

        JavaDStream<OrderWide> orderWideDStream = orderJoinDStream.mapPartitions(new FlatMapFunction<Iterator<Tuple2<Long, Tuple2<Optional<OrderInfo>, Optional<OrderDetail>>>>, OrderWide>() {
            @Override
            public Iterator<OrderWide> call(Iterator<Tuple2<Long, Tuple2<Optional<OrderInfo>, Optional<OrderDetail>>>> tuple2Iterator) throws Exception {
                Jedis jedis = MyRedisUtils.getJedisFromPool();
                List<OrderWide> orderWides = new ArrayList<>();
                while(tuple2Iterator.hasNext()) {
                    Tuple2<Long, Tuple2<Optional<OrderInfo>, Optional<OrderDetail>>> tuple2All = tuple2Iterator.next();
                    Tuple2<Optional<OrderInfo>, Optional<OrderDetail>> tuple2 = tuple2All._2;
                    Optional<OrderInfo> orderInfoOptional = tuple2._1;
                    Optional<OrderDetail> orderDetailOptional = tuple2._2;
                    //orderInfo有， orderDetail有
                    if(orderInfoOptional.isPresent()) {
                        //取出orderInfo
                        OrderInfo orderInfo = orderInfoOptional.get();
                        if(orderDetailOptional.isPresent()) {
                            //取出orderDetail
                            OrderDetail orderDetail = orderDetailOptional.get();
                            //组装成orderWide
                            OrderWide orderWide = new OrderWide(orderInfo, orderDetail);
                            orderWides.add(orderWide);
                        }
                        //orderInfo有，orderDetail没有

                        //orderInfo写缓存
                        // 类型:  string
                        // key :   ORDERJOIN:ORDER_INFO:ID
                        // value :  json
                        // 写入API:  set
                        // 读取API:  get
                        // 是否过期: 24小时
                        String redisOrderInfoKey = "ORDERJOIN:ORDER_INFO:" + orderInfo.getId();
                        jedis.setex(redisOrderInfoKey, 24 * 3600,JSON.toJSONString(orderInfo));
                        //orderInfo读缓存
                        String redisOrderDetailKey = "ORDERJOIN:ORDER_DETAIL:" + orderInfo.getId();
                        Set<String> orderDetailsSet = jedis.smembers(redisOrderDetailKey);
                        if(orderDetailsSet != null && !orderDetailsSet.isEmpty()) {
                            Iterator it = orderDetailsSet.iterator();
                            while(it.hasNext()) {
                                OrderDetail orderDetail = JSON.parseObject(String.valueOf(it.next()),OrderDetail.class);
                                //组装成orderWide
                                OrderWide orderWide = new OrderWide(orderInfo, orderDetail);
                                orderWides.add(orderWide);
                            }
                        }
                    }else {
                        //orderInfo没有， orderDetail有
                        OrderDetail orderDetail = orderDetailOptional.get();
                        String redisOrderInfoKey = "ORDERJOIN:ORDER_INFO:" + orderDetail.getOrder_id();
                        //读缓存
                        String orderInfoJson = jedis.get(redisOrderInfoKey);
                        if(orderInfoJson != null && orderInfoJson.length() > 0) {
                            OrderInfo orderInfo = JSON.parseObject(orderInfoJson, OrderInfo.class);
                            //组装成orderWide
                            OrderWide orderWide = new OrderWide(orderInfo, orderDetail);
                            //加入到结果集中
                            orderWides.add(orderWide);
                        } else {
                            //写缓存
                            // 类型:   set
                            // key :   ORDERJOIN:ORDER_DETAIL:ORDER_ID
                            // value :  json, json ....
                            // 写入API: sadd
                            // 读取API: smembers
                            // 是否过期: 24小时
                            String redisOrderDetailKey = "ORDERJOIN:ORDER_DETAIL:" + orderDetail.getOrder_id();
                            jedis.sadd(redisOrderDetailKey, JSON.toJSONString(orderDetail));
                            jedis.expire(redisOrderDetailKey, 24 * 3600);

                        }
                    }
                }
                jedis.close();
                return orderWides.iterator();
            }
        });

        //写入ES
        //1. 索引分割， 通过索引模板控制mapping setting aliases
        //2. 使用工具类将数据写入ES
        orderWideDStream.foreachRDD(new VoidFunction<JavaRDD<OrderWide>>() {
            @Override
            public void call(JavaRDD<OrderWide> orderWideJavaRDD) throws Exception {
                orderWideJavaRDD.foreachPartition(new VoidFunction<Iterator<OrderWide>>() {
                    @Override
                    public void call(Iterator<OrderWide> orderWideIterator) throws Exception {
                        List<Tuple2<String, Object>> docs = new ArrayList<>();
                        String date = null;
                        while(orderWideIterator.hasNext()) {
                            OrderWide orderWide = orderWideIterator.next();
                            if(date == null) {
                                date = orderWide.getCreate_date();
                            }
                            Tuple2<String, Object> tuple2 = new Tuple2<>(orderWide.getDetail_id().toString(), orderWide.toObject());
                            docs.add(tuple2);
                        }
                        if(!docs.isEmpty()) {
                            String indexName = "gmall_order_wide_1018_" + date;
                            MyEsUtils.bulkSave(indexName , docs);
                        }
                    }
                });
                MyOffsetsUtils.saveOffset(orderInfoTopicName , orderInfoGroup , orderInfoOffsetRanges);
                MyOffsetsUtils.saveOffset(orderDetailTopicName , orderDetailGroup ,orderDetailOffsetRanges);
            }
        });
    }
}
